﻿//========================================================= 
/**@file ThreadTask.h
 * @brief 基于线程的任务调度
 * 
 * @date 2018-10-12   11:35:14
 * @author Zhyioo 
 * @version 1.0
 */ 
//--------------------------------------------------------- 
#ifndef _LIBZHOUYB_THREADTASK_H_
#define _LIBZHOUYB_THREADTASK_H_
//--------------------------------------------------------- 
#include "../../include/Base.h"

#include <condition_variable>
#include <mutex>
#include <thread>
#include <memory>
#include <vector>
#include <list>
//--------------------------------------------------------- 
namespace zhou_yb {
namespace extension {
namespace ability {
//--------------------------------------------------------- 
/**
 * @brief 线程任务池
 * @date 2018-10-12 11:51:29
 */
template<class TTaskParam>
class ThreadTask
{
public:
    typedef TTaskParam ParamType;
protected:
    std::mutex _mutex;
    std::condition_variable_any _condition;
    std::list<TTaskParam> _queue;
    std::vector<std::shared_ptr<std::thread> > _pool;
    bool _is_running;

    virtual void OnTask(ParamType& param) = 0;
    virtual void OnQuit(ParamType& param) {}

    void run()
    {
        while(_is_running)
        {
            TTaskParam param;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                if(_queue.size() > 0)
                {
                    param = _queue.front();
                    _queue.pop_front();
                }
                else
                {
                    _condition.wait(lock);
                    continue;
                }
            }
            OnTask(param);
        }
    }
public:
    ThreadTask() : _is_running(false) {}
    virtual ~ThreadTask() { Quit(); }

    void Init(size_t thread_count = 1)
    {
        _is_running = true;
        for(size_t i = 0; i < thread_count; ++i)
        {
            std::shared_ptr<std::thread> t = std::make_shared<std::thread>(std::bind(&ThreadTask::run, this));
            _pool.push_back(t);
        }
    }
    inline bool IsReady() const
    {
        return _pool.size() > 0;
    }
    virtual void Push(const ParamType& param)
    {
        // 如任务未开始,则默认以单线程的方式启动
        if(!IsReady()) Init();

        std::unique_lock<std::mutex> lock(_mutex);
        _queue.push_back(param);
        _condition.notify_one();
    }
    virtual void Quit()
    {
        if(!_is_running)
            return;

        _is_running = false;
        _condition.notify_all();

        std::vector<std::shared_ptr<std::thread> >::iterator itr;
        for(itr = _pool.begin(); itr != _pool.end(); ++itr)
        {
            (*itr)->join();
        }
        _pool.clear();
        
        std::list<ParamType>::iterator paramitr;
        for(paramitr = _queue.begin(); paramitr != _queue.end(); ++paramitr)
        {
            OnQuit(*paramitr);
        }
        _queue.clear();
    }
};
//--------------------------------------------------------- 
} // namespace ability 
} // namespace extension
} // namespace zhou_yb
//--------------------------------------------------------- 
#endif // _LIBZHOUYB_THREADTASK_H_
//========================================================= 